package com.ng.biz.service.impl;

import java.util.List;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.jfinal.kit.StrKit;
import com.jfinal.plugin.activerecord.Db;
import com.jfinal.plugin.activerecord.Page;
import com.jfinal.plugin.activerecord.Record;
import com.ng.biz.model.domain.DataLogInput;
import com.ng.biz.model.domain.DataTask;
import com.ng.biz.model.domain.Datasource;
import com.ng.biz.service.IDataErrorLog;
import com.ng.biz.service.IDataFlow;
import com.ng.biz.service.IDataLog;
import com.ng.biz.service.IDataTask;
import com.ng.common.enums.DataLogStatusEnum;
import com.ng.common.model.dto.ParsedSql;
import com.ng.common.util.ParsedSqlUtil;
import lombok.extern.slf4j.Slf4j;
import com.ng.common.util.DateUtil;
import org.springframework.data.redis.core.RedisTemplate;
import java.util.Date;

@Slf4j
@Service
public class DataTaskImpl implements IDataTask{
	
	@Autowired
	private IDataFlow dataFlowService;
	
	@Autowired
	private RedisTemplate<String, Object> redisTemplate;
	
    @Autowired
    private IDataLog dataLogService;
    
	@Autowired
	private IDataErrorLog dataErrorLogService;

	@Override
	public void executeTask(String dataLogId,String taskId) throws Exception {
		DataTask dataTask=DataTask.dao.findById(taskId);
		if(dataTask!=null) {
			try {
				// 解析输入配置
				JSONObject input = JSON.parseObject(dataTask.getInput());
				String datasourceId = input.getString("datasourceId");
				JSONObject content = JSON.parseObject(dataTask.getContent());
				executeOLTPTask(dataLogId,dataTask,datasourceId,content);
			} catch(Exception e) {
				log.error("执行任务失败: {}", taskId, e);
				dataErrorLogService.insertOrUpdateErrorLog(dataTask.getId(), dataTask.getName(), "OLTP", "error", e.getMessage());
				throw e;
			}
		}
	}

	/**
	 * 执行OLTP类型任务
	 * @throws Exception 
	 */
	private void executeOLTPTask(String dataLogId,DataTask dataTask,String datasourceId, JSONObject content) throws Exception {
		String sql = content.getString("sql");
		JSONArray srcParams = content.getJSONArray("params");
		
		List<JSONObject> params = null;
		if(srcParams==null || srcParams.isEmpty()) {
			 params=new LinkedList<JSONObject>();
		}
		else {
			params = JSON.parseObject(  JSONObject.toJSONString(srcParams), new TypeReference<List<JSONObject>>() {});
		}

		String isPage = content.getString("isPage");
		Integer limit = content.getInteger("limit");
		
		int threads=5;
		if(content.containsKey("threads")) {
			threads = content.getInteger("threads");
		}
		
		try {
			// 获取数据源
			Datasource datasource = Datasource.dao.findById(datasourceId);
			if(datasource == null) {
				log.error("数据源不存在: {}", datasourceId);
				throw new Exception(String.format("数据源不存在: %s", datasourceId));
			}
			
			if("1".equals(isPage) && limit != null) {
				// 分页处理
				processWithPagination(dataLogId,dataTask,datasourceId,sql, params, limit,threads);
			} else {
				// 一次性处理
				processWithoutPagination(dataLogId,dataTask,datasourceId,sql, params);
			}
			
		} catch(Exception e) {
			log.error("{} 执行SQL失败: {}", sql, e);
			throw e;
		}
		
	}
	
	/**
	 * 分页处理数据
	 * @throws Exception 
	 */
	private void processWithPagination(String dataLogId,DataTask dataTask,String datasourceId,String sql, List<JSONObject> fieds, int pageSize,int threads) throws Exception {
		// 解析SQL和参数
		ParsedSql parsedSql = ParsedSqlUtil.parseSql(sql, fieds,new JSONObject());
		
		if(parsedSql == null) {
			throw new Exception(String.format("%sSQL解析失败,请检查是否有误.", dataTask.getName()));
		}
		
		
		dataLogService.buildUpdateDataLog(dataLogId)
		.setInfo(String.format("%s创建线程池，线程数量:%s.", dataTask.getName(),threads))
		.update(); 
		
		int pageNumber = 1;
		// 创建线程池
		ExecutorService executorService = Executors.newFixedThreadPool(threads);
		List<CompletableFuture<Void>> futures = new ArrayList<>();
		
		dataLogService.buildUpdateDataLog(dataLogId)
		.setInfo(String.format("%s开始执任务", dataTask.getName()))
		.update(); 
		
		// 处理第一页
		Page<Record> page = Db.use(datasourceId).paginate(pageNumber, pageSize, parsedSql.getSql(),"", parsedSql.getParams().toArray());
		
		if(page.getList()==null || page.getList().size()<=0) {
			dataLogService.buildUpdateDataLog(dataLogId)
			.setWarning(1)
			.setInfo(String.format("%s未查询到数据.", dataTask.getName()))
			.update(); 
			return;
		}
		
		
		DataLogInput dataLogInput=dataLogService.createDataLogInput(dataLogId, 1, page.getList().size());
		
		processPageData(dataLogId,dataLogInput.getId(),dataTask,page.getList(),1);
		
		if(page.getTotalPage() > 1) {
			
			for(int cur = 2; cur <= page.getTotalPage(); cur++) {
				final int currentPage = cur;
				CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
					try {
						Page<Record> pageData = Db.use(datasourceId).paginate(currentPage, pageSize, parsedSql.getSql(),"",parsedSql.getParams().toArray());
						DataLogInput dataLogInputS=dataLogService.createDataLogInput(dataLogId, currentPage, pageData.getList().size());
						processPageData(dataLogId,dataLogInputS.getId(),dataTask,pageData.getList(),currentPage);
					} catch(Exception e) {
						log.error("处理第{}页数据失败: {}", currentPage, e.getMessage());
					}
				}, executorService);
				
				futures.add(future);
			}
			
			// 等待所有任务完成
			try {
				dataLogService.buildUpdateDataLog(dataLogId)
				.setInfo(String.format("%s完成任务创建完成，等待futures:%s结束.", dataTask.getName(),futures.size()))
				.update(); 
				for(Future<?> future : futures) {
					future.get();
				}
				
				List<DataLogInput> dataLogInputError=DataLogInput.dao.find("select id from "+DataLogInput.TABLE_NAME+" where log_id=? and status=?" ,dataLogId,DataLogStatusEnum.Error.getType());
				List<DataLogInput> dataLogInputWarning=DataLogInput.dao.find("select id from "+DataLogInput.TABLE_NAME+" where log_id=? and status=?" ,dataLogId,DataLogStatusEnum.Error.getType());
				
				dataLogService.buildUpdateDataLog(dataLogId)
				.setError(dataLogInputError != null?dataLogInputError.size():0)
				.setWarning(dataLogInputWarning != null?dataLogInputWarning.size():0)
				.setInfo(String.format("%s完成任务创建等待，执行结束:%s.", dataTask.getName(),threads))
				.update(); 
				
				
				
			} catch(Exception e) {
				log.error("等待任务完成超时: {}", e.getMessage());
				throw new Exception(String.format("%s分页处理数据异常：%s.", dataTask.getName(),e.getMessage()));
			} finally {
				executorService.shutdown();
				try {
					if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
						executorService.shutdownNow();
					}
				} catch (InterruptedException e) {
					executorService.shutdownNow();
				}
			}
		}
	}
	
	/**
	 * 不分页处理数据
	 * @throws Exception 
	 */
	private void processWithoutPagination(String dataLogId,DataTask dataTask,String datasourceId,String sql, List<JSONObject> fields) throws Exception {
		
		// 解析SQL和参数
		ParsedSql parsedSql = ParsedSqlUtil.parseSql(sql, fields,new JSONObject());
		if(parsedSql == null) {
			throw new Exception(String.format("%sSQL解析失败,请检查是否有误.", dataTask.getName()));
		}
		
		dataLogService.buildUpdateDataLog(dataLogId)
		.setInfo(String.format("%s开始执任务", dataTask.getName()))
		.update(); 
		
		List<Record> records = Db.use(datasourceId).find(parsedSql.getSql(), parsedSql.getParams().toArray());
		
		if(records==null || records.size()<=0) {
			dataLogService.buildUpdateDataLog(dataLogId)
			.setWarning(1)
			.setInfo(String.format("%s未查询到数据.", dataTask.getName()))
			.update(); 
			return;
		}
		
		DataLogInput dataLogInput=dataLogService.createDataLogInput(dataLogId, 1, records.size());

		processPageData(dataLogId,dataLogInput.getId(),dataTask,records,1);
		
		List<DataLogInput> dataLogInputError=DataLogInput.dao.find("select id from "+DataLogInput.TABLE_NAME+" where log_id=? and status=?" ,dataLogId,DataLogStatusEnum.Error.getType());
		List<DataLogInput> dataLogInputWarning=DataLogInput.dao.find("select id from "+DataLogInput.TABLE_NAME+" where log_id=? and status=?" ,dataLogId,DataLogStatusEnum.Error.getType());

		dataLogService.buildUpdateDataLog(dataLogId)
		.setInfo(String.format("%s任务执行结束.", dataTask.getName()))
		.setError(dataLogInputError != null?dataLogInputError.size():0)
		.setWarning(dataLogInputWarning != null?dataLogInputWarning.size():0)
		.update(); 
		
	}
	
	
	/**
	 * 处理数据
	 */
	private void processPageData(String dataLogId,String dataLogInputId,DataTask dataTask,List<Record> records,int currentPage) {
		// TODO: 实现数据处理逻辑
		// 1. 数据清洗
		// 2. 数据转换
		// 3. 数据输出
		//log.info("处理数据条数: {}", records.size());
		
		if(records==null || records.size()<=0) {
			return;
		}
		
		dataLogService.
		buildUpdateDataLogInput(dataLogInputId).
		setStatus(DataLogStatusEnum.Normal.getType()).
		setInfo(String.format("%s开始处理第%s页数据.", dataTask.getName(),currentPage)).
		update();
		
		JSONArray standardFields = JSONArray.parseArray(dataTask.getStandardFields());
		
		List<JSONObject> standards=new LinkedList<JSONObject>();
		for(Record record: records) {
			standards.add(convertInput(record,standardFields));
		}
		
		JSONArray jsonArray=JSONArray.parseArray(dataTask.getOutput());
		if(jsonArray==null || jsonArray.isEmpty()) {
			dataLogService.
			buildUpdateDataLogInput(dataLogInputId).
			setStatus(DataLogStatusEnum.Warning.getType()).
			setInfo(String.format("%s未配置输出源，丢弃第%s页数据.", dataTask.getName(),currentPage)).
			update();
			return;
		}
		try {
			dataFlowService.processOutputData(dataLogId, dataLogInputId,jsonArray, standards);
			
			dataLogService.
			buildUpdateDataLogInput(dataLogInputId).
			setStatus(DataLogStatusEnum.Success.getType()).
			setInfo(String.format("%s第%s页数据处理结束.", dataTask.getName(),currentPage)).
			setTotal(records.size()).
			setEndTime(new Date(System.currentTimeMillis())).
			update();
		} catch (Exception e) {
			log.error("processPageData异常:{}",e);
			dataLogService.
			buildUpdateDataLogInput(dataLogInputId).
			setStatus(DataLogStatusEnum.Error.getType()).
			setEndTime(new Date(System.currentTimeMillis())).
			setInfo(String.format("%s处理第%s页数据异常:%s.", dataTask.getName(),currentPage,e.getMessage())).
			update();
		}
		
	}
	
	/**
	 * 转换输入数据
	 * @param data 输入数据
	 * @param standardFields 标准字段配置
	 * @return 转换后的标准数据
	 */
	private JSONObject convertInput(Record data, JSONArray standardFields) {
		if(data == null || standardFields == null) {
			return null;
		}
		
		JSONObject result = new JSONObject();
		try {
			for(int i = 0; i < standardFields.size(); i++) {
				JSONObject field = standardFields.getJSONObject(i);
				
				// 1. 输入字段配置
				//String inputFieldName = field.getString("inputFieldName");
				String inputFieldIdentifier = field.getString("inputFieldIdentifier");
				String inputFieldType = field.getString("inputFieldType");
				String inputFieldDateFormatSrc = field.getString("inputFieldDateFormatSrc");
				String inputFieldDateFormat = field.getString("inputFieldDateFormat");
				String inputFieldFormat = field.getString("inputFieldFormat");
				
				// 2. 输出字段配置
				//String outputFieldName = field.getString("outputFieldName");
				String outputFieldIdentifier = field.getString("outputFieldIdentifier");
				String outputFieldType = field.getString("outputFieldType");
				String outputFieldFormat = field.getString("outputFieldFormat");
				String outputFieldDefaultValue = field.getString("outputFieldDefaultValue");
				String outputFieldDecimalPoint = field.getString("outputFieldDecimalPoint");
				String outputValueFromInput = field.getString("outputValueFromInput");
				
				// 3. 处理字段名和标识符相同性
				if("1".equals(field.getString("sameFieldName"))) {
					//outputFieldName = inputFieldName;
				}
				if("1".equals(field.getString("sameFieldIdentifier"))) {
					outputFieldIdentifier = inputFieldIdentifier;
				}
				
				// 4. 获取输入值并处理
				Object inputValue = data.get(inputFieldIdentifier);
				if(inputValue != null) {
					// 4.1 处理日期类型
					if("dateString".equals(inputFieldType)) {
						String dateFormat = "select".equals(inputFieldDateFormatSrc) ? 
							inputFieldDateFormat : inputFieldFormat;
						Date date = DateUtil.dateStrToDate(inputValue.toString(), dateFormat);
						if(date != null) {
							inputValue = date;
						}
					}
				}
				
				// 5. 根据输出值来源处理
				Object finalValue = null;
				switch(outputValueFromInput) {
					case "input" -> finalValue = inputValue;
					
					case "constant" -> finalValue = outputFieldDefaultValue;
					
					case "curSystemDate" -> {
						if(List.of("dateString", "long", "timestampString").contains(outputFieldType)) {
							finalValue = new Date();
						}
					}
					
					case "OLTP" -> {
						String datasourceId = field.getString("outputValueDatasourceId");
						String sql = field.getString("outputValueSql");
						if(StrKit.notBlank(sql)) {
							// 替换SQL中的参数
							String columnName = inputFieldIdentifier.contains(".") ? 
								inputFieldIdentifier.substring(inputFieldIdentifier.lastIndexOf(".") + 1) : 
								inputFieldIdentifier;
							sql = sql.replace("${" + columnName + "}", 
								inputValue != null ? inputValue.toString() : "");
							finalValue = Db.use(datasourceId).queryStr(sql);
						}
					}
					
					case "redis" -> {
						String redisKey = field.getString("outputValueRedisKey");
						if(StrKit.notBlank(redisKey)) {
							finalValue = redisTemplate.opsForValue().get(redisKey);
						}
					}
				}
				
				// 6. 处理默认值
				if(finalValue == null && StrKit.notBlank(outputFieldDefaultValue)) {
					finalValue = outputFieldDefaultValue;
				}
				
				// 7. 类型转换和输出
				if(finalValue != null && StrKit.notBlank(outputFieldIdentifier)) {
					// 7.1 根据输出字段类型进行转换
					switch(outputFieldType.toLowerCase()) {
						case "string" -> finalValue = finalValue.toString();
						
						case "int" -> finalValue = Integer.parseInt(finalValue.toString());
						
						case "float", "double" -> {
							if(StrKit.notBlank(outputFieldDecimalPoint)) {
								int decimalPoints = Integer.parseInt(outputFieldDecimalPoint);
								double value = Double.parseDouble(finalValue.toString());
								finalValue = String.format("%." + decimalPoints + "f", value);
							}
						}
						
						case "boolean" -> finalValue = Boolean.parseBoolean(finalValue.toString());
						
						case "datestring" -> {
							if(finalValue instanceof Date) {
								finalValue = DateUtil.getFormatDate((Date)finalValue, 
									StrKit.notBlank(outputFieldFormat) ? outputFieldFormat : "yyyy-MM-dd HH:mm:ss");
							}
						}
					}
					
					result.put(outputFieldIdentifier, finalValue);
				}
			}
			
			return result;
		} catch(Exception e) {
			log.error("Convert input data failed", e);
			return null;
		}
	}
}
